package com.markhsiu.minimq.zookeeper;

import org.apache.zookeeper.KeeperException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class ZkUtils {
	
	private static Logger logger = LoggerFactory.getLogger(ZkUtils.class);
	public static final String ROOT = "/minimq";
	public static final String ROOT_BROKER = "/minimq/broker";
	public static final String ROOT_CONSUMER = "/minimq/consumer";
	 
	private static final ZkClient ZK = ZkClient.instance();
	
	public static void getAllBroker(){	
		 
	}	
	
	public static void registerCusuomerTopic(String consumerID,String topic){	
		try {
			String path = ZkUtils.ROOT_CONSUMER+"/"+consumerID+"/"+topic;
			if(ZK.exists(path)){
				logger.info("consumer：{} 下已存在 {}",consumerID, topic);
				return;
			}
			logger.info("consumer:{}下注册 {}", consumerID,topic);
			ZK.create(path, false);			
		} catch (InterruptedException | KeeperException e) {
			e.printStackTrace();
		}
	}	
	
	public static void clearCusuomerTopic(String consumerID,String topic){	
		try {
			String path = ZkUtils.ROOT_CONSUMER+"/"+consumerID+"/"+topic;
			if(!ZK.exists(path)){
				logger.info("consumer：{} 下不存在 {}",consumerID, topic);
				return;
			}
			logger.info("consumer:{}下删除 {}", consumerID,topic);
			ZK.delete(path);			
		} catch (InterruptedException | KeeperException e) {
			e.printStackTrace();
		}
	}
	
	public static void registerBrokerTopic(String brokerID,String topic){	
		if(!ZkUtils.checkBroker()){
			ZkUtils.initMQ();
		}
		
		try {
			String path = ZkUtils.ROOT_BROKER+"/"+brokerID+"/"+topic;
			if(ZK.exists(path)){
				logger.info("broker：{} 下已存在 {}",brokerID, topic);
				return;
			}
			logger.info("broker:{}下注册 {}", brokerID,topic);
			ZK.create(path, false);				
		} catch (InterruptedException | KeeperException e) {
			e.printStackTrace();
		}
	}	
	
	public static void registerCusuomer(String consumerID){	
		if(!ZkUtils.checkBroker()){
			ZkUtils.initMQ();
		}
		
		try {
			String path = ZkUtils.ROOT_CONSUMER+"/"+consumerID;
			if(ZK.exists(path)){
				logger.info("consumerID 已存在 ：{}", consumerID);
				return;
			}
			logger.info("注册Consumer ：{}", consumerID);
			ZK.create(path, false);			
		} catch (InterruptedException | KeeperException e) {
			e.printStackTrace();
		}
	}	
	
	
	
	public static void registerBroker(String brokerID,String address){	
		if(!ZkUtils.checkBroker()){
			ZkUtils.initMQ();
		}
		
		try {
			String path = ZkUtils.ROOT_BROKER+"/"+brokerID;
			if(ZK.exists(path)){
				logger.info("brokerID 已存在 ：{}", brokerID);
				return;
			}
			logger.info("注册broker ：{}", brokerID);
			ZK.create(path, false);			
			ZK.setData(path, address);
		} catch (InterruptedException | KeeperException e) {
			e.printStackTrace();
		}
	}	

	
	private static boolean checkBroker(){
		try {
			return ZK.exists(ZkUtils.ROOT_BROKER);
		} catch (InterruptedException e) {
			e.printStackTrace();
		} catch (KeeperException e) {
			e.printStackTrace();
		}
		return false;
	}
	
	private static void initMQ(){
		try {
			ZK.create(ROOT, true);
			ZK.create(ROOT_BROKER, true);
			ZK.create(ROOT_CONSUMER, true);
			logger.info("初始化MQ zookeeper 节点");
		} catch (KeeperException e) {
			e.printStackTrace();
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}
	
	public static ZkClient getZkClient(){
		return ZK;
	}
	
	public static void close() throws InterruptedException{
		 ZK.ZKClose();
	}
}
